package com.shujia.stream

import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo4Structured {
  def main(args: Array[String]): Unit = {


    val spark: SparkSession = SparkSession.builder()
      .config("spark.sql.shuffle.partitions", 2)
      .master("local[5]")
      .appName("stru")
      .getOrCreate()


    /**
      *
      * 读取socket
      */

    val linesDF: DataFrame = spark
      .readStream
      .format("socket")
      .option("host", "master")
      .option("port", 9999)
      .load()


    linesDF.createOrReplaceTempView("words")


    val countDF: DataFrame = spark.sql(
      """
        |select value as word,count(1) as c from words group by value
        |
      """.stripMargin)


    countDF
      .writeStream
      .outputMode(OutputMode.Complete()) //流模式
      .format("console") //输出到控制台
    .start() //启动一个流
    .awaitTermination() //等待关闭


  /*  import spark.implicits._
    import org.apache.spark.sql.functions._

    linesDF
      .groupBy($"value")
      .agg(count($"value") as "c")
      .createOrReplaceTempView("count") //动态表


    spark
      .table("count") //获取表
      .writeStream
      .outputMode(OutputMode.Complete()) //流模式
      .format("console") //输出到控制台
      .start() //启动一个流
      .awaitTermination() //等待关闭
*/

  }
}
